package com.lsp.kafkaprovider.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class SendMessageController {

    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;

    @RequestMapping("/send/simple")
    public String sendSimpleMsg(@RequestParam String msg){
        kafkaTemplate.send("springboottopic", msg);
        return "success";
    }

    @RequestMapping("/send/batch")
    public String sendBatchMsg(@RequestParam String msg){
        for (int i =0 ; i < 23 ; i++){
            kafkaTemplate.send("springboottopic", msg+i);
        }
        return "success";
    }

    @RequestMapping("/send/async")
    public String sendAsyncMsg(@RequestParam String msg){
        kafkaTemplate.send("springboottopic", msg).addCallback(success   -> {
            // 消息发送到的topic
            String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            System.out.println("发送消息成功:" + topic + "，分区：" + partition + "，偏移量：" + offset);
        },failure ->{
            System.out.println("发送消息失败:" + failure.getMessage());
        });
        return "executed";
    }



    @RequestMapping("/send/transaction")
    public String sendTransactionMsg(@RequestParam String msg){
        // 声明事务：operations函数报错，消息就不会发出去。
        kafkaTemplate.executeInTransaction(operations ->{
            //数据发往kafka
            operations.send("springboottopic", msg);
            //模拟后续业务处理发生了异常
            throw new RuntimeException("fail");
        });
        return "executed";
    }

}
